package com.icehand.canal.client.transfer;

import com.alibaba.otter.canal.client.CanalConnector;
import com.alibaba.otter.canal.protocol.CanalEntry;
import com.alibaba.otter.canal.protocol.Message;
import com.google.common.collect.Lists;
import com.icehand.canal.annotation.HandlePoint;
import com.icehand.canal.client.Point;
import com.icehand.canal.config.CanalConfig;
import com.icehand.canal.handler.CanalHandler;
import lombok.extern.slf4j.Slf4j;

import java.util.Collections;
import java.util.List;

/**
 * @author icehand
 */
@Slf4j
public abstract class BaseMessageTransponder implements MessageTransponder {
    protected final CanalConnector canalConnector;

    protected final CanalConfig.Instance instance;

    protected final String destination;

    protected final List<CanalHandler> handlers = Lists.newArrayList();

    protected final List<Point> points = Lists.newArrayList();

    private volatile  boolean running = true;

    public BaseMessageTransponder(CanalConnector canalConnector, CanalConfig.Instance instance,
                                  String destination, List<CanalHandler> handlers, List<Point> points){
        this.canalConnector = canalConnector;
        this.instance = instance;
        this.destination = destination;
        if(null != handlers) {
            this.handlers.addAll(handlers);
        }
        if(null != points) {
            this.points.addAll(points);
        }
    }

    @Override
    public void run() {
        int errorCnt = instance.getRetryCount();
        long interval = instance.getAcquireInterval();
        String threadName = Thread.currentThread().getName();
        while(this.running && !Thread.currentThread().isInterrupted()) {
            try {
                Message message = canalConnector.getWithoutAck(instance.getBatchSize());
                long batchId = message.getId();
                int size = message.getEntries().size();
                if(log.isDebugEnabled()) {
                    log.debug("【{}】: Get message from canal server，size【{}】",threadName,size);
                }

                if(batchId == -1 || size == 0){
                    Thread.sleep(interval);
                } else {
                    distributeMessage(message);
                }

                canalConnector.ack(batchId);
                if(log.isDebugEnabled()) {
                    log.debug("【{}】ack message batchId 【{}】",threadName,batchId);
                }
            } catch (InterruptedException e) {
                errorCnt -- ;
                canalConnector.rollback();
            } finally {
                if(errorCnt < 1) {
                    stop();
                    log.info("【{}】topping the client", Thread.currentThread().getName());
                }
            }
        }

        stop();
        log.info("{}: client stopped",Thread.currentThread().getName());
    }

    protected abstract void distributeMessage(Message message);

    void stop(){
        this.running = false;
    }
}
